import { AppTask } from "../../app.task.class";
import { BehaviorSubject, groupBy, Observable, Subject } from "rxjs";
import mqtt from "mqtt";
import { SRAlertTopicHandlerTask } from "./sr.alert.topic.handler.task";
import { uploadFile$ } from "../operator/sr.operator";
import { V3proSubjectMessage } from "../../v3pro-task/model/v3pro.model";

/**
 * SR错误信息推送任务
 * 1. 订阅错误信息
 * 2. 推送错误信息
 * 3. 部分错误信息需要先推送文件
 */
export class SrAlertMessageServer extends AppTask {
  constructor(
    private __mqttSubject: BehaviorSubject<mqtt.MqttClient | undefined>,
    private __srAlertMessagePublishSubject: Subject<V3proSubjectMessage>,
  ) {
    super();
  }

  /**
   * 暂停任务
   */
  __pause(): Promise<boolean> {
    return Promise.resolve(false);
  }

  /**
   * 恢复任务
   */
  __restore(): Promise<boolean> {
    return Promise.resolve(true);
  }

  /**
   * 运行任务
   * 1. 添加 mqtt 变化监听
   * 2. 添加错误信息监听
   */
  async __run(): Promise<void> {
    this.__srAlertMessagePublishSubject
      .pipe(
        groupBy((message) => message.srMqttTopic),
        uploadFile$,
      )
      .subscribe(this.__groupEventHandler.bind(this));
    await this.__await();
  }

  // 当前分组任务
  private __groupMessageTasks: SRAlertTopicHandlerTask[] = [];

  /**
   * 管道分组处理
   * @param group$
   */
  async __groupEventHandler(group$: Observable<V3proSubjectMessage>) {
    this.__groupMessageTasks.push(new SRAlertTopicHandlerTask(this.__mqttSubject, group$));
  }

  /**
   * 停止任务
   */
  async __stop(): Promise<boolean> {
    this.__groupMessageTasks.forEach((task) => task.stop());
    return Promise.resolve(true);
  }
}
